package com.wpre.vampiremq.mqclient;

import com.wpre.vampiremq.common.*;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: LianBao
 * Date: 2025-01-19
 * Time: 20:43
 */
public class Connection {

    private Socket socket = null;

    //需要管理多个Channel
    private ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<String, Channel>();

    private ExecutorService callBackPoll;

    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;

    public Connection(String host, int port) throws IOException {
        socket = new Socket(host,port);
        inputStream = socket.getInputStream();
        outputStream = socket.getOutputStream();
        dataInputStream = new DataInputStream(inputStream);
        dataOutputStream = new DataOutputStream(outputStream);
        callBackPoll = Executors.newFixedThreadPool(4);

        //创建一个扫描线程,这个线程负责从socket中读取响应数据,把这个响应数据交给对应的channel负责处理
        Thread t = new Thread(() -> {
            try {
                while (!socket.isClosed()) {
                    Response response = readResponse();
                    dispatchResponse(response);
                }
            } catch (SocketException socketException) {

            } catch (IOException | ClassNotFoundException e) {

            }
        });
        t.start();

    }
    //关闭Connection,释放资源
    public void close(){
        callBackPoll.shutdownNow();
        channelMap.clear();
        try {
            inputStream.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            outputStream.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        try {
            socket.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    //分别处理 当前响应是针对控制请求的响应还是服务器推送的消息
    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException {
        if (response.getType() == 0xc) {
            //服务器推送的消息
            SubscribeReturns returns = (SubscribeReturns) BinaryTool.fromBytes(response.getPayload());
            Channel channel = channelMap.get(returns.getChannelId());
            if (channel == null) {
                throw new MQException("Channel不存在");
            }
            callBackPoll.submit(() -> {
                try {
                    channel.getConsumer().handleDelivery(returns.getConsumerTag(), returns.getBasicProperties(), returns.getBody());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            });
        } else {
            //针对控制请求的响应进行处理
            BasicReturns returns = (BasicReturns) BinaryTool.fromBytes(response.getPayload());
            //根据channelId找到channel对象,
            Channel channel = channelMap.get(returns.getChannelId());
            if (channel == null) {
                throw new MQException("Channel不存在");
            }
            channel.putReturns(returns);
        }

    }

    //发送请求
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType());
        dataOutputStream.writeInt(request.getLength());
        dataOutputStream.write(request.getPayload());
        dataOutputStream.flush();
    }

    //读取响应
    public Response readResponse() throws IOException {
        Response response = new Response();
        response.setType(dataInputStream.readInt());
        response.setLength(dataInputStream.readInt());
        byte[] payload = new byte[response.getLength()];
        int read = dataInputStream.read(payload);
        if (read != response.getLength()) {
            throw new IOException("读取的响应数据不完整");
        }
        response.setPayload(payload);

        return response;
    }

    //在Connection中创建出一个Channel
    public Channel createChannel() throws IOException {

        String channelId = "C-" + UUID.randomUUID().toString();

        Channel channel = new Channel(channelId, this);
        //加入哈希表中
        channelMap.put(channelId, channel);

        boolean ok = channel.createChannel();
        if (!ok) {
            //如果服务器这这边创建失败,要把channel从哈希表中删除
            channelMap.remove(channelId);
            return null;
        }

        return channel;
    }


}
